package com.shujia.flink.source;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.Getter;
import lombok.Setter;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

public class Demo4MySQLSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //使用自定义source
        DataStream<Student> studentDS = env.addSource(new MySQlSource());

        DataStream<Tuple2<String, Integer>> kvDS = studentDS
                .map(stu -> Tuple2.of(stu.getClazz(), 1), Types.TUPLE(Types.STRING, Types.INT));

        DataStream<Tuple2<String, Integer>> countDS = kvDS.keyBy(kv -> kv.f0).sum(1);

        countDS.print();

        env.execute();


    }
}

/**
 * 自定义source 读取mysql中的数据
 * 练习：自定义source 读取redis中的数据
 */
class MySQlSource implements SourceFunction<Student> {
    /**
     * run方法再任务启动的时候执行一次
     */
    @Override
    public void run(SourceContext<Student> ctx) throws Exception {
        //1、加载驱动
        Class.forName("com.mysql.jdbc.Driver");
        //2、创建数据库连接
        Connection con = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata29", "root", "123456");
        //3、编写sql查询数据
        PreparedStatement stat = con.prepareStatement("select * from student");
        //4、执行查询
        ResultSet resultSet = stat.executeQuery();
        //5、解析数据
        while (resultSet.next()) {
            int id = resultSet.getInt("id");
            String name = resultSet.getString("name");
            int age = resultSet.getInt("age");
            String gender = resultSet.getString("gender");
            String clazz = resultSet.getString("clazz");

            //将数据发送到下游
            ctx.collect(new Student(id, name, age, gender, clazz));
        }

        //4、关闭数据库连接
        stat.close();
        con.close();
    }

    @Override
    public void cancel() {
    }
}


/**
 * lombok插件作用：可以再代码编译时动态给代码增加方法（相当于scala中的case class）
 *
 * @Getter
 * @Setter
 * @AllArgsConstructor
 */
@Data
@AllArgsConstructor
class Student {
    private Integer id;
    private String name;
    private Integer age;
    private String gender;
    private String clazz;
}

